package com.enhanced.mqtt.client;

import com.enhanced.mqtt.manager.EmqxServerManager;
import org.eclipse.paho.client.mqttv3.MqttException;

import java.util.Scanner;
import java.util.concurrent.TimeUnit;

/**
 * 高级MQTT客户端示例
 * 演示EMQX的高级特性：
 * 1. 共享订阅
 * 2. 保留消息
 * 3. 延迟消息
 * 4. 遗嘱消息
 * 5. 用户名密码认证
 */
public class AdvancedMqttClient {
    
    private static final String BROKER_URL = "tcp://192.168.3.234:1883";
    private static final String TOPIC = "test/topic";
    private static final String SHARED_GROUP = "group1";
    private static final String WILL_TOPIC = "clients/disconnected";
    private static final int QOS = 1;
    
    public static void main(String[] args) {
        try {
            runDemo(new Scanner(System.in));
        } catch (Exception e) {
            System.out.println("MQTT错误: " + e.getMessage());
            e.printStackTrace();
        }
    }
    
    /**
     * 运行演示，使用外部传入的Scanner
     * @param scanner 已经打开的Scanner
     */
    public static void runDemo(Scanner scanner) throws Exception {
        System.out.println("EMQX高级特性演示");
        System.out.println("请选择要演示的功能：");
        System.out.println("1. 共享订阅演示");
        System.out.println("2. 保留消息演示");
        System.out.println("3. 延迟消息演示");
        System.out.println("4. 遗嘱消息演示");
        System.out.println("5. 认证演示");
        System.out.print("请输入选项(1-5): ");
        
        int choice = scanner.nextInt();
        scanner.nextLine(); // 消费换行符
        
        try {
            switch (choice) {
                case 1:
                    demonstrateSharedSubscription();
                    break;
                case 2:
                    demonstrateRetainedMessages();
                    break;
                case 3:
                    demonstrateDelayedMessages();
                    break;
                case 4:
                    demonstrateWillMessages();
                    break;
                case 5:
                    demonstrateAuthentication();
                    break;
                default:
                    System.out.println("无效的选择！");
            }
        } catch (Exception e) {
            System.out.println("演示过程中发生错误: " + e.getMessage());
            e.printStackTrace();
            throw e;
        }
        // 不关闭scanner，由调用者负责关闭
    }
    
    /**
     * 演示共享订阅功能
     * 创建多个客户端，共享同一个主题的订阅
     */
    private static void demonstrateSharedSubscription() throws MqttException, InterruptedException {
        System.out.println("=== 共享订阅演示 ===");
        System.out.println("启动两个客户端，共享同一个主题的订阅");
        
        // 创建第一个客户端
        EmqxServerManager client1 = EmqxServerManager.createDefault(BROKER_URL);
        client1.connectWithDefaultOptions();
        client1.subscribeShared(SHARED_GROUP, TOPIC, QOS);
        System.out.println("客户端1已连接并订阅共享主题");
        
        // 创建第二个客户端
        EmqxServerManager client2 = EmqxServerManager.createDefault(BROKER_URL);
        client2.connectWithDefaultOptions();
        client2.subscribeShared(SHARED_GROUP, TOPIC, QOS);
        System.out.println("客户端2已连接并订阅共享主题");
        
        // 创建发布者客户端
        EmqxServerManager publisher = EmqxServerManager.createDefault(BROKER_URL);
        publisher.connectWithDefaultOptions();
        
        // 发布10条消息
        System.out.println("开始发布10条消息，这些消息将被负载均衡到两个客户端");
        for (int i = 1; i <= 10; i++) {
            String message = "共享订阅测试消息 #" + i;
            publisher.publish(TOPIC, message, QOS, false);
            TimeUnit.MILLISECONDS.sleep(500);
        }
        
        // 关闭连接
        TimeUnit.SECONDS.sleep(2);
        client1.close();
        client2.close();
        publisher.close();
        System.out.println("共享订阅演示结束");
    }
    
    /**
     * 演示保留消息功能
     * 发布一条保留消息，然后新客户端订阅时会立即收到
     */
    private static void demonstrateRetainedMessages() throws MqttException, InterruptedException {
        System.out.println("=== 保留消息演示 ===");
        
        // 创建发布者客户端
        EmqxServerManager publisher = EmqxServerManager.createDefault(BROKER_URL);
        publisher.connectWithDefaultOptions();
        
        // 发布保留消息
        String retainedTopic = "retained/message";
        System.out.println("发布保留消息到主题: " + retainedTopic);
        publisher.publish(retainedTopic, "这是一条保留消息，新订阅者会立即收到", QOS, true);
        
        // 等待一段时间
        System.out.println("等待2秒，然后创建新的订阅者...");
        TimeUnit.SECONDS.sleep(2);
        
        // 创建一个新的客户端订阅同一主题
        EmqxServerManager subscriber = EmqxServerManager.createDefault(BROKER_URL);
        subscriber.connectWithDefaultOptions();
        System.out.println("新客户端订阅主题，应该立即收到保留消息");
        subscriber.subscribe(retainedTopic, QOS);
        
        // 等待接收消息
        TimeUnit.SECONDS.sleep(5);
        
        // 清除保留消息
        System.out.println("清除保留消息（发布空消息并设置retained=true）");
        publisher.publish(retainedTopic, "", QOS, true);
        
        // 关闭连接
        publisher.close();
        subscriber.close();
        System.out.println("保留消息演示结束");
    }
    
    /**
     * 演示延迟消息功能
     * 发布几条不同延迟的消息，观察消息到达的顺序
     */
    private static void demonstrateDelayedMessages() throws MqttException, InterruptedException {
        System.out.println("=== 延迟消息演示 ===");
        
        // 创建客户端
        EmqxServerManager client = EmqxServerManager.createDefault(BROKER_URL);
        client.connectWithDefaultOptions();
        client.subscribe(TOPIC, QOS);
        
        // 发布不同延迟的消息
        System.out.println("发布3条不同延迟的消息:");
        client.publishDelayed(TOPIC, "5秒后到达的消息", QOS, 5);
        System.out.println("已发送5秒延迟消息");
        
        client.publishDelayed(TOPIC, "2秒后到达的消息", QOS, 2);
        System.out.println("已发送2秒延迟消息");
        
        client.publish(TOPIC, "立即到达的消息", QOS, false);
        System.out.println("已发送即时消息");
        
        // 等待所有消息到达
        System.out.println("等待所有延迟消息到达...");

        // 关闭连接
        System.out.println("延迟消息演示结束");
    }
    
    /**
     * 演示遗嘱消息功能
     * 设置遗嘱消息，然后异常断开连接，观察遗嘱消息的发送
     */
    private static void demonstrateWillMessages() throws MqttException, InterruptedException {
        System.out.println("=== 遗嘱消息演示 ===");
        
        // 创建一个监听遗嘱主题的客户端
        EmqxServerManager listener = EmqxServerManager.createDefault(BROKER_URL);
        listener.connectWithDefaultOptions();
        listener.subscribe(WILL_TOPIC, QOS);
        System.out.println("监听客户端已订阅遗嘱主题: " + WILL_TOPIC);
        
        // 创建带遗嘱消息的客户端
        EmqxServerManager clientWithWill = EmqxServerManager.createDefault(BROKER_URL);
        clientWithWill.connect(null, null, WILL_TOPIC, "客户端异常断开连接", true);
        System.out.println("带遗嘱消息的客户端已连接");
        
        // 等待一会儿
        System.out.println("等待3秒后模拟客户端异常断开...");
        TimeUnit.SECONDS.sleep(3);
        
        // 模拟客户端异常断开（不调用正常的断开方法）
        System.out.println("模拟客户端异常断开");
        clientWithWill.getClient().close(true);
        
        // 等待遗嘱消息被接收
        System.out.println("等待遗嘱消息被接收...");
        TimeUnit.SECONDS.sleep(5);
        
        // 关闭监听客户端
        listener.close();
        System.out.println("遗嘱消息演示结束");
    }
    
    /**
     * 演示用户名密码认证功能
     * 尝试使用正确和错误的凭据连接
     */
    private static void demonstrateAuthentication() throws MqttException {
        System.out.println("=== 认证演示 ===");
        System.out.println("注意：此演示需要EMQX配置了用户名密码认证");
        
        Scanner scanner = new Scanner(System.in);
        System.out.print("请输入用户名: ");
        String username = scanner.nextLine();
        System.out.print("请输入密码: ");
        String password = scanner.nextLine();
        
        try {
            // 使用用户名密码连接
            EmqxServerManager client = EmqxServerManager.createDefault(BROKER_URL);
            client.connect(username, password, null, null, true);
            System.out.println("认证成功！");
            
            // 测试发布和订阅
            client.subscribe(TOPIC, QOS);
            client.publish(TOPIC, "认证成功后的测试消息", QOS, false);
            
            // 关闭连接
            client.close();
        } catch (MqttException e) {
            System.out.println("认证失败: " + e.getMessage());
        }
        
        System.out.println("认证演示结束");
        scanner.close();
    }
} 